package qdu.edu.cn.storm.topology;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.LocalDRPC;
import org.apache.storm.StormSubmitter;
import org.apache.storm.drpc.LinearDRPCTopologyBuilder;
import org.apache.storm.tuple.Fields;
import qdu.edu.cn.storm.bolt.SplitWordsBolt;
import qdu.edu.cn.storm.redis.write.RedisStoreRequestListBolt;

import static qdu.edu.cn.storm.ConstantFinalString.FUNCTION;
import static qdu.edu.cn.storm.ConstantFinalString.Fields.USER_ID;

public class DRPCSearchWebTopology {
    public static void main(String[] args) throws Exception {
        String topologyName = "D-RPC-SearchWeb-topology";

        LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder(FUNCTION);
        builder.addBolt(new SplitWordsBolt(), 4).shuffleGrouping();
        builder.addBolt(new RedisStoreRequestListBolt(), 2).fieldsGrouping(new Fields(USER_ID));

        Config config = new Config();

        if (args == null || args.length == 0) {
            LocalDRPC localDRPC = new LocalDRPC();
            LocalCluster localCluster = new LocalCluster();

            localCluster.submitTopology(topologyName, config, builder.createLocalTopology(localDRPC));
            for (int i = 0; i < 10; i++) {
                String execute = localDRPC.execute(FUNCTION, "鞋子，靴子，连衣裙，连衣裙");
                System.out.println(execute);
            }
            localCluster.shutdown();
            localDRPC.shutdown();
        } else {
            StormSubmitter.submitTopologyWithProgressBar(topologyName, config, builder.createRemoteTopology());
        }

    }
}
